package org.pupil.proxy.cglib;

import io.netty.channel.ChannelFuture;
import io.netty.channel.DefaultEventLoop;
import io.netty.util.concurrent.DefaultPromise;
import net.sf.cglib.proxy.MethodInterceptor;
import net.sf.cglib.proxy.MethodProxy;
import org.pupil.annotation.RpcReference;
import org.pupil.common.*;
import org.pupil.common.constants.*;
import org.pupil.register.RegistryFactory;
import org.pupil.router.LoadBalancer;
import org.pupil.router.LoadBalancerFactory;
import org.pupil.socket.codec.MsgHeader;
import org.pupil.socket.codec.RpcProtocol;
import org.pupil.socket.codec.RpcRequest;
import org.pupil.socket.codec.RpcResponse;

import java.lang.reflect.Method;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @Description: CGLIB实现的动态代理类，代理服务请求调用，简化客户端的调用
 * @Author: pupil
 * @Date: 2024/07/10 下午 4:28
 */
public class CgLibProxy implements MethodInterceptor {

    private final String serviceName; // 服务名

    private final String version; // 版本


    private final long time; // 异步等待时间

    private final TimeUnit timeUnit; // 异步等待时间单位

    public CgLibProxy(Class clazz) {
        this.serviceName = clazz.getName();
        final RpcReference rpcService = (RpcReference) clazz.getAnnotation(RpcReference.class);
        version = rpcService.version();
        time = rpcService.time();
        timeUnit = rpcService.timeUnit();
    }

    /**
     * @param o           被代理的对象（需要增强的对象）
     * @param method      被拦截的方法（需要增强的方法）
     * @param objects     方法入参
     * @param methodProxy 用于调用原始方法
     */
    @Override
    public Object intercept(Object o, Method method, Object[] objects, MethodProxy methodProxy) throws Throwable {
        final RpcProtocol<RpcRequest> rpcProtocol = new RpcProtocol<>();
        // 构建消息头
        MsgHeader header = new MsgHeader();
        header.setMagic(ProtocolConstants.MAGIC);
        header.setVersion(ProtocolConstants.VERSION);
        header.setMsgType((byte) MsgType.REQUEST.ordinal());
        long requestId = RpcRequestHolder.getRequestId();
        header.setRequestId(requestId);
        header.setStatus((byte) 0x1);
        final byte[] serialization = RpcSerialization.JSON.name.getBytes();
        header.setSerializationLen(serialization.length);
        header.setSerializations(serialization);
        // 设置请求头
        rpcProtocol.setMsgHeader(header);

        // 构建请求消息
        final RpcRequest rpcRequest = new RpcRequest();
        rpcRequest.setClassName(method.getDeclaringClass().getName());
        rpcRequest.setMethodName(method.getName());
        rpcRequest.setMethodCode(method.hashCode());
        rpcRequest.setServiceVersion(version);
        // 判断参数是否为空
        if (null != objects && objects.length > 0) {
            rpcRequest.setParameterTypes(objects[0].getClass());
            rpcRequest.setParameter(objects[0]);
        }
        // 设置请求体
        rpcProtocol.setBody(rpcRequest);

        // 获取服务提供方资源定位符集合
        final List<URL> urls = RegistryFactory.get(Register.ZOOKEEPER).discoveries(serviceName, version);
        if (urls.isEmpty()) {
            throw new Exception("无服务可用:" + serviceName);
        }

        // 通过路由层获取服务提供方资源定位符
        final LoadBalancer loadBalancer = LoadBalancerFactory.get(LoadBalance.Round);
        final URL url = loadBalancer.select(urls);

        // 获取ChannelFuture
        final ChannelFuture channelFuture = Cache.CHANNEL_FUTURE_MAP.get(new Host(url.getIP(), url.getPort()));
        // 发送消息
        channelFuture.channel().writeAndFlush(rpcProtocol);

        // 异步监听器
        RpcFuture<RpcResponse> future = new RpcFuture<>(new DefaultPromise<>(new DefaultEventLoop()), time);
        // 根据请求id绑定异步监听器
        RpcRequestHolder.REQUEST_MAP.put(requestId, future);
        // 获取返回的响应消息，设置超时时间的阻塞等待
        RpcResponse rpcResponse = future.getPromise().sync().get(future.getTimeOut(),timeUnit);
        if (rpcResponse.getException() != null) {
            throw rpcResponse.getException();
        }
        // 返回响应消息的数据
        return rpcResponse.getData();
    }
}
